package com.xjj.sso.client.rabbitmq;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import javax.servlet.http.HttpSession;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import com.xjj.sso.client.SSOConstants;
import com.xjj.sso.client.filter.SSOClientFilter;
import com.xjj.sso.client.session.SessionHandle;

public class QueueConsumer extends EndPoint implements Runnable, Consumer{

	
	
	public QueueConsumer(String endPointName) throws IOException, TimeoutException{
        super(endPointName);       
    }
     
    public void run() {
        try {
        	
        	channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            String queueName = channel.queueDeclare().getQueue();
            channel.queueBind(queueName, EXCHANGE_NAME, endPointName);
        	
            //start consuming messages. Auto acknowledge messages.
        	//第二个参数：是否自动应答 
            channel.basicConsume(queueName,true,this);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
 
    /**
     * Called when consumer is registered.
     */
    public void handleConsumeOk(String consumerTag) {
        System.out.println("Consumer "+consumerTag +" registered");    
    }
 
  
    /**
     * Called when new message is available.
     */
    public void handleDelivery(String consumerTag, Envelope env,
            BasicProperties props, byte[] body) throws IOException {
    	
    	//收到退出消息
        System.out.println("consumer Message : "+ new String(body) + " received.");
        
        MqMessage msg = JSON.parseObject(new String(body), MqMessage.class);
        
        if(null != msg && "logout".equals(msg.getTopic()) && SSOClientFilter.TICKET_SESSION_CACHE.containsKey(msg.getContent()))
		{
			HttpSession sess = SSOClientFilter.TICKET_SESSION_CACHE.get(msg.getContent());
			
			
			if(null != sess)
			{
				SessionHandle handle;
				try {
					handle = (SessionHandle) (Class.forName(SSOConstants.SSO_CLIENT_SESSIONHANDLE).newInstance());
					handle.ssoLogout(msg.getContent(),sess);
				} catch (Exception e) {
					e.printStackTrace();
				}
				//sso不再做处理，由各业务系统处理
//				SSOFilter.SESSION_TICKET_CACHE.remove(sess.getId());
//				try {
//					sess.invalidate();
//				} catch (Exception e) {
//					
//				}
//				SSOFilter.TICKET_SESSION_CACHE.remove(msg.getContent());
			}
		}
    }
    
    
	@Override
	public void handleCancelOk(String consumerTag) {
		
	}

	@Override
	public void handleCancel(String consumerTag) throws IOException {
		
	}


	@Override
	public void handleShutdownSignal(String consumerTag,
			ShutdownSignalException sig) {
		
	}

	@Override
	public void handleRecoverOk(String consumerTag) {
		
	}
    
    
}
